/*
 */

package jas.rpc.server.cluster;

import jas.http.server.HttpServer;
import jas.rpc.NonRemoteMethodException;
import jas.rpc.Remote;
import jas.rpc.server.ObjectIdBank;
import jas.rpc.server.RpcServerFactory;
import java.lang.reflect.InvocationTargetException;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.util.Util;
import jas.rpc.RpcRequest;
import jas.rpc.server.RpcListener;
import jas.rpc.server.RpcListenerImpl;
import jas.rpc.server.RpcServerObjectInputStream;
import jas.rpc.server.RpcServerObjectOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;

/**
 *
 *
 */
public class JGroupRpcServer extends ReceiverAdapter implements Runnable, RpcListener {
    
    private HttpServer httpServer;
    private ObjectIdBank ids;
    private JChannel channel;
    private JGroupStateFactory factory;
    private String groupName;
    private RpcListener childListener;
    
    public JGroupRpcServer(String groupName, JGroupStateFactory factory) {
        this.groupName = groupName;
        this.factory = factory;
        this.childListener = new RpcListenerImpl();
    }
    
    public void run() {
        try {
            // Use the sequencer total ordering from the JAR file.
            channel = new JChannel(getClass().getResource("/sequencer.xml"));
            channel.setReceiver(this);
            channel.connect(groupName);
            if (!channel.getState(null, 10000)) {
                ids = factory.createState();
                System.out.println("I am the first in the cluster, creating new bank.");
            } else {
                // setState() was called, setting the "ids" fields
                System.out.println("Received bank state from cluster.");
            }
            httpServer = new HttpServer(new RpcServerFactory(ids, this));
            httpServer.run();
        } catch (ChannelException ex) {
            ex.printStackTrace();
        } finally {
            channel.close();
        }
    }
    
    @Override
    public byte[] getState() {
        System.out.println("Sending state.");
        try {
            return Util.objectToByteBuffer(ids);
        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not serialize state.", ex);
        }
    }
    
    @Override
    public void setState(byte[] bytes) {
        System.out.println("Receiving state.");
        try {
            ids = (ObjectIdBank) Util.objectFromByteBuffer(bytes);
            System.out.println("Received state: " + ids);
        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not deserialize state.", ex);
        }
    }
    
    @Override
    public void receive(Message msg) {
        // Ignore our own messages
        if (msg.getSrc() == channel.getLocalAddress())
            return;
        
        System.out.println("Received message!");
        try {
            RpcRequest request = (RpcRequest) deserialize(msg.getBuffer()); // Read-only so don't clone the buffer.
            System.out.println("request = " + request);
            Object o = childListener.call(request);
            if (o instanceof Remote) {
                ids.getId(o); // Assign an ID to the returned remote object: IMPORTANT!!!!
            }
        } catch (NoSuchMethodException ex) {
            ex.printStackTrace();
            
        } catch (InvocationTargetException ex) {
            ex.printStackTrace();
            
        } catch (NonRemoteMethodException ex) {
            ex.printStackTrace();
            
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
    
    public Object call(RpcRequest request) throws NoSuchMethodException, InvocationTargetException, NonRemoteMethodException {
        Method m = request.objectToCall.getClass().getMethod(request.methodName, request.classes);
        
        // If method is read-only, just call it and return without notifying other servers
        if (m.isAnnotationPresent(ReadOnly.class)) {
            return childListener.call(request);
        }
        
        try {
            System.out.println("call!");
            
            Message msg = new Message(null, null, serialize(request));
            channel.send(msg);
            
            System.out.println("Message sent!");
        } catch (ChannelNotConnectedException ex) {
            ex.printStackTrace();
            
        } catch (ChannelClosedException ex) {
            ex.printStackTrace();
            
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return childListener.call(request);
    }
    
    private byte[] serialize(Object o) {
        try {
            ByteArrayOutputStream buffer = new ByteArrayOutputStream();
            RpcServerObjectOutputStream stream = new RpcServerObjectOutputStream(buffer, ids);
            //stream.setDebug(true);
            stream.writeObject(o);
            return buffer.toByteArray();
        } catch (IOException ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not serialize request.", ex);
        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not serialize request.", ex);
        }
    }
    
    private Object deserialize(byte[] bytes) {
        try {
            ByteArrayInputStream buffer = new ByteArrayInputStream(bytes);
            RpcServerObjectInputStream stream = new RpcServerObjectInputStream(buffer, ids);
            return stream.readObject();
        } catch (IOException ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not deserialize request.", ex);
        } catch (ClassNotFoundException ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not deserialize request.", ex);
        } catch (Exception ex) {
            ex.printStackTrace();
            throw new RuntimeException("Could not deserialize request.", ex);
        }
    }
    
}
